package com.cuit.filter;

import com.cuit.constants.Constants;
import com.cuit.enums.CompressTypeEnum;
import com.cuit.enums.RpcErrorMessageEnum;
import com.cuit.enums.SerializationTypeEnum;
import com.cuit.exception.RpcException;
import com.cuit.extension.Activate;
import com.cuit.extension.ExtensionLoader;
import com.cuit.factory.SingleFactory;
import com.cuit.registry.ServiceDiscovery;
import com.cuit.remoting.constants.RpcConstants;
import com.cuit.remoting.dto.RpcMessage;
import com.cuit.remoting.dto.RpcRequest;
import com.cuit.remoting.dto.RpcResponse;
import com.cuit.remoting.transport.netty.client.ChannelProvider;
import com.cuit.remoting.transport.netty.client.UnProcessRequests;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Activate(group = Constants.CONSUMER, order = 99)
public class InvokeFilter implements Filter {

    private final ServiceDiscovery serviceDiscovery;
    private final ChannelProvider channelProvider;
    private final UnProcessRequests unProcessRequests;

    public InvokeFilter() {
        this.unProcessRequests = SingleFactory.getSingle(UnProcessRequests.class);
        this.channelProvider = SingleFactory.getSingle(ChannelProvider.class);
        this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension("zk");
    }

    @Override
    public void doFilter(FilterContext fIlterContext) throws Exception{
        invoke(fIlterContext);
    }

    public void invoke(FilterContext fIlterContext) throws Exception{
        RpcRequest rpcRequest = fIlterContext.getRpcRequest();
        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);
        Channel channel = channelProvider.get(inetSocketAddress);
        if(channel.isActive()){
            CompletableFuture<RpcResponse<Object>> resultFuture = fIlterContext.getResultFuture();
            unProcessRequests.put(rpcRequest.getRequestId(),resultFuture);
            RpcMessage rpcMessage = RpcMessage.builder().data(rpcRequest)
                    .codec(SerializationTypeEnum.PROTOSTUFF.getCode())
                    .compress(CompressTypeEnum.GZIP.getCode())
                    .messageType(RpcConstants.REQUEST_TYPE).build();
            channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future->{
                if(future.isSuccess()){
                    log.info("客服端发送消息{}",rpcMessage);
                }else{
//                    future.channel().close();
                    resultFuture.completeExceptionally(future.cause());
                    log.error("发送消息失败[{}]",future.cause());
                    throw new RpcException(RpcErrorMessageEnum.SERVICE_INVOCATION_FAILED);
                }
            });
        }else{
            throw new IllegalStateException();
        }
    }
}
